package org.bikelab.sephiroce;

import java.io.IOException;
//import java.util.StringTokenizer;

//import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
//import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
//import org.apache.hadoop.mapreduce.Reducer;
//import org.apache.hadoop.mapreduce.Reducer.Context;
import org.bikelab.se3.rdf.reader.NoLineToParseException;
import org.bikelab.se3.rdf.reader.RdfReaderFactory;
import org.bikelab.se3.rdf.reader.WrongNumberOfSegmentsException;
//import org.bikelab.test.WordCount.IntSumReducer;
//import org.bikelab.test.WordCount.TokenizerMapper;
import org.semanticweb.yars.nx.Node;

public class ModifySPOCtoSPOS {
	public static class SPOCParserMapper extends Mapper<Object, Text, Text, Text>{
	    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
	    	try {
				Node[] quad = RdfReaderFactory.parseNqLine(value.toString());
				context.write(new Text(quad[0].toN3()), new Text(quad[1].toN3()+" "+quad[2].toN3()+" "+quad[0].toN3()));
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (WrongNumberOfSegmentsException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (NoLineToParseException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	    }
	 }
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
	    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
	    if (otherArgs.length != 2) {
	      System.err.println("Usage: wordcount <in> <out>");
	      System.exit(2);
	    }
	    Job job = new Job(conf, "word count");
	    job.setJarByClass(ModifySPOCtoSPOS.class);
	    job.setMapperClass(SPOCParserMapper.class);
	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(Text.class);
	    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
	    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
	    System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}
